package de.mmis.core.infrastructure;

import static de.mmis.core.base.infrastructure.InfrastructureEvent.InfrastructureEventType.*;
import static de.mmis.core.config.Config.*;
import static de.mmis.core.infrastructure.Message.MessageType.*;

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.InterfaceAddress;
import java.net.NetworkInterface;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.net.UnknownHostException;
import java.util.Arrays;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Random;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;

import org.apache.log4j.Logger;

import de.mmis.core.base.Pair;
import de.mmis.core.base.abstracttree.InnerNode;
import de.mmis.core.base.abstracttree.LeafNode;
import de.mmis.core.base.abstracttree.Tree;
import de.mmis.core.base.abstracttree.TreeParserException;
import de.mmis.core.base.abstracttree.serializer.DeserializableException;
import de.mmis.core.base.infrastructure.InfrastructureEvent;
import de.mmis.core.base.infrastructure.InfrastructureException;
import de.mmis.core.base.infrastructure.PublishedDeviceDescription;
import de.mmis.core.infrastructure.Message.MessageType;
import de.mmis.core.sexpression.SExpressionParser;
import de.mmis.core.sexpression.SExpressionWriter;

/**
 * <p>
 * The RemoteServer communicates via UDP broadcasts with other RemoteServers. It
 * manages all machine-local devices.
 * </p>
 * 
 * <p>
 * There is only one instance of this class on one machine.
 * </p>
 * 
 * @author Martin Nyolt
 * 
 */
class HostServer {

	public final static Logger LOGGER = Logger.getLogger(HostServer.class);

	final static ThreadGroup INFRASTRUCTURE_THREAD_GROUP = new ThreadGroup(
			"Infrastructure Threads");
	private final static ThreadGroup CLIENT_THREAD_GROUP = new ThreadGroup(
			INFRASTRUCTURE_THREAD_GROUP, "local client threads");

	private class ParseMessage implements Runnable {

		private final byte[] b;
		private final InetAddress a;

		public ParseMessage(final byte[] b, final InetAddress a) {
			this.b = b;
			this.a = a;
		}

		@Override
		public void run() {
			parseBroadcastPacket(b, a);
		}

	}

	class RemoteServerReceiveThread extends Thread {

		public RemoteServerReceiveThread() {
			super(INFRASTRUCTURE_THREAD_GROUP, "RemoteServerThread");
			setDaemon(true);
		}

		@Override
		public void run() {
			byte[] buf;
			try {
				buf = new byte[socket.getReceiveBufferSize()];
			} catch (SocketException e1) {
				buf = new byte[1500];
			}

			boolean wasError = false;

			ExecutorService pool = Executors.newFixedThreadPool(1,
					new ThreadFactory() {
						int i = 0;

						@Override
						public Thread newThread(Runnable r) {
							Thread result = new Thread(
									INFRASTRUCTURE_THREAD_GROUP, r);
							result.setName("RemoteServerParseThread-" + (++i));
							result.setDaemon(true);
							return result;
						}
					});

			while (running) {
				DatagramPacket packet = new DatagramPacket(buf, buf.length);
				try {
					socket.receive(packet);
					if (!running)
						break;
					wasError = false;

					final byte[] b = Arrays.copyOf(packet.getData(),
							packet.getLength());
					final InetAddress a = packet.getAddress();
					pool.execute(new ParseMessage(b, a));

				} catch (IOException e) {
					if (!running)
						// we wanted to get this exception
						break;

					LOGGER.error("Could not receive UDP packet", e);

					if (wasError) {
						LOGGER.error("Two errors in a line - try to close the socket and create a new one");
						socket.close();
						try {
							socket = createSocket();
						} catch (SocketException e1) {
							LOGGER.fatal(
									"Unable to create a new socket - giving up",
									e1);
							HostServer.this.stop();
							break;
						}
						wasError = false;
					} else
						wasError = true;
				}
			}
		}
	}

	class LocalServerClientThread extends Thread {

		private Client client;

		public LocalServerClientThread(Client client) {
			super(CLIENT_THREAD_GROUP, "LocalServerClientThread-"
					+ client.socket.getRemoteSocketAddress());
			setDaemon(true);
			this.client = client;
		}

		@Override
		public void run() {
			// send hello message to client
			sendClientMessage(Message.hello(getBindAddress()), client);

			while (running) {
				Tree message = null;
				try {
					message = SExpressionParser.parse(client.input, false);
				} catch (TreeParserException e) {
					if (!client.socket.isClosed()
							&& !e.unexpectedEndOfStreamOccurred())
						LOGGER.error("Can't parse S-Expression from client "
								+ client, e);
					break;
				} catch (IOException e) {
					if (!client.socket.isClosed())
						LOGGER.error("Can't read from socket of client "
								+ client, e);
					break;
				}

				if (client.socket.isClosed())
					break;

				if (message == null)
					continue;

				try {
					parseClientMessage(message.getAs(Message.class), client);
				} catch (InfrastructureException e) {
					LOGGER.error("Can't parse message " + message
							+ " from client " + client, e);
				} catch (DeserializableException e) {
					LOGGER.error("can't deserialize message " + message, e);
				}
			}

			synchronized (clients) {
				try {
					client.socket.close();
				} catch (IOException e) {
					// ignore
				}
				client.socket = null;
				clients.remove(client);
			}

			LOGGER.debug("removing client " + client);

			// remove all existing devices of the client (dirty)
			try {
				removeDevices(client, false);
			} catch (InfrastructureException e) {
				LOGGER.warn("Can't remove devices of client " + client);
			}
		}
	}

	class LocalServerListenThread extends Thread {

		public LocalServerListenThread() {
			super(INFRASTRUCTURE_THREAD_GROUP, "LocalServerListenThread");
			setDaemon(true);
		}

		@Override
		public void run() {
			while (running) {
				Socket clientSocket;
				try {
					clientSocket = clientServerSocket.accept();
				} catch (IOException e) {
					if (running) {
						LOGGER.error("Can't accept new clients", e);
						// if we can't accept new clients the socket seems to be
						// broken
						running = false;
						HostServer.this.stop();
					}
					break;
				}

				try {
					Client client = new Client(clientSocket);
					synchronized (clients) {
						clients.add(client);
					}
					LOGGER.debug("new client " + client);
					new LocalServerClientThread(client).start();
				} catch (IOException e) {
					LOGGER.error("Can't create client from socket "
							+ clientSocket);
				}
			}

		}

	}

	private final LinkedList<Client> clients = new LinkedList<Client>();

	private ServerSocket clientServerSocket;

	/**
	 * To map an answer (via the ID) to the appropriate client
	 */
	private final HashMap<Tree, Client> queries = new HashMap<Tree, Client>();

	private final Random random = new Random();
	/** ID to identify this machine */
	private Tree machineID;

	private final LinkedList<Pair<Client, PublishedDeviceDescription>> devices = new LinkedList<Pair<Client, PublishedDeviceDescription>>();

	private boolean running;
	private InetAddress bindAddress = null;
	private DatagramSocket socket;
	private DatagramSocket sendSocket;

	private InetAddress broadcastAddress;

	final static LeafNode heartbeatMessage = new LeafNode("heartbeat");
	private final Timer heartbeatTimer = new Timer("Infrastructure Heartbeat",
			true);

	static List<InetAddress> getPossibleBindAddresses() {
		List<InetAddress> result = new LinkedList<InetAddress>();

		Enumeration<NetworkInterface> nis;
		try {
			nis = NetworkInterface.getNetworkInterfaces();
		} catch (SocketException e) {
			LOGGER.warn("can't get netwrk interfaces", e);
			return result;
		}

		while (nis.hasMoreElements()) {
			NetworkInterface ni = nis.nextElement();
			try {
				// avoid loopback interfaces
				if (ni.isLoopback()) {
					continue;
				}
			} catch (SocketException e) {
				// ignore
			}

			List<InterfaceAddress> as = ni.getInterfaceAddresses();
			for (InterfaceAddress a : as)
				if (a.getBroadcast() != null)
					result.add(a.getAddress());
		}

		return result;
	}

	/**
	 * Starts the RemoteServer in a seperate Thread.
	 * 
	 * @param first
	 *            true if this is a new machine, false if it's taking over
	 * @param clean
	 *            false if the old VM did not announced it wants to stop
	 * @param bindAddress
	 *            The address to which the {@link HostServer} should be bound
	 * 
	 * @throws SocketException
	 * 
	 * @throws InfrastructureException
	 *             when the "new machine" or "migrated" message could not be
	 *             sent.
	 */
	void start(boolean first, boolean clean, InetAddress bindAddress)
			throws InfrastructureException, IOException {
		if (bindAddress != null) {
			try {
				broadcastAddress = InetAddress.getByName("255.255.255.255");
			} catch (UnknownHostException e) {
				// should not happen
				LOGGER.error("could not create InetAddress 255.255.255.255", e);
			}
			this.bindAddress = bindAddress;
		} else {
			for (InetAddress a : getPossibleBindAddresses()) {
				this.bindAddress = a;
				if (a.getHostAddress().startsWith("139.30.")) {
					LOGGER.trace("Found preferrable network interface " + a);
					break;
				}
			}
		}

		if (this.bindAddress == null) {
			LOGGER.debug("No none-loopback network interface found");

			NetworkInterface loopback = null;
			Enumeration<NetworkInterface> nis = NetworkInterface
					.getNetworkInterfaces();
			while (nis.hasMoreElements()) {
				NetworkInterface ni = nis.nextElement();
				if (ni.isLoopback()) {
					loopback = ni;
					break;
				}
			}

			if (loopback == null)
				throw new InfrastructureException("No loopback interface found");

			InterfaceAddress a = loopback.getInterfaceAddresses().get(0);
			this.bindAddress = a.getAddress();
			broadcastAddress = a.getBroadcast();
		}

		// always set broadcast to 255.255.255.255
		try {
			broadcastAddress = InetAddress.getByName("255.255.255.255");
		} catch (UnknownHostException e) {
			// should not happen
			LOGGER.error("could not create InetAddress 255.255.255.255", e);
		}

		clientServerSocket = new ServerSocket(INFRASTRUCTURE_PORT);

		int id = clientServerSocket.getInetAddress().getHostAddress()
				.hashCode()
				+ random.nextInt();
		machineID = new LeafNode(Integer.toHexString(id));

		LOGGER.info("bindAddress: " + this.bindAddress + " - broadcast: "
				+ broadcastAddress);

		try {
			sendSocket = new DatagramSocket(0, this.bindAddress);
			socket = createSocket();
		} catch (SocketException e) {
			try {
				clientServerSocket.close();
			} catch (IOException e1) {
				// ignore
			}
			clientServerSocket = null;
			if (sendSocket != null) {
				sendSocket.close();
				sendSocket = null;
			}

			throw e;
		}

		running = true;

		new LocalServerListenThread().start();
		new RemoteServerReceiveThread().start();

		InfrastructureEvent event;
		if (first) {
			event = new InfrastructureEvent(NEW_MACHINE);
		} else {
			event = new InfrastructureEvent(MIGRATED);
			event.clean = clean;
		}
		try {
			sendBroadcastMessage(Message.event(event));
		} catch (InfrastructureException e) {
			LOGGER.error("Can't send event " + event, e);
		}

		// init heartbeat

		String heartbeatPeriod_ = System.getProperty(
				"Infrastructure.heartbeatPeriod", "300");

		long heartbeatPeriod;
		try {
			heartbeatPeriod = Long.parseLong(heartbeatPeriod_);
		} catch (NumberFormatException e) {
			heartbeatPeriod = 300;
			LOGGER.error("Heartbeat period " + heartbeatPeriod_
					+ " is not a valid number, defaulting to "
					+ heartbeatPeriod);
		}

		LOGGER.debug("Sending heartbeat message every " + heartbeatPeriod
				+ " seconds");

		// send heartbeat message
		heartbeatTimer.schedule(new TimerTask() {
			@Override
			public void run() {
				try {
					sendBroadcastMessage(Message.event(new InfrastructureEvent(
							HEARTBEAT)));
				} catch (InfrastructureException e) {
					LOGGER.error("Could not send heartbeat message", e);
				}
			}
		}, heartbeatPeriod * 100, heartbeatPeriod * 1000);
	}

	private DatagramSocket createSocket() throws SocketException {
		DatagramSocket result = new DatagramSocket(INFRASTRUCTURE_PORT);
		result.setBroadcast(true);

		return result;
	}

	private InetAddress getBindAddress() {
		return bindAddress;
	}

	/**
	 * stop the server. We don't say "device removed" here because another VM
	 * could set up a new Server.
	 */
	void stop() {
		if (!running)
			return;

		// avoid parsing messages sent here
		running = false;
		heartbeatTimer.cancel();

		// announce that we want to shut down
		InfrastructureEvent event = new InfrastructureEvent(
				SHUT_DOWN_ANNOUNCEMENT);
		try {
			sendBroadcastMessage(Message.event(event));
		} catch (InfrastructureException e) {
			LOGGER.warn("Error sending shut-down message", e);
		}

		// also send this message to VMs on this machine
		try {
			event.address = InetAddress.getLocalHost();
			sendClientMessage(Message.event(event));
		} catch (UnknownHostException e) {
			// should not happen
			LOGGER.warn("Can't get InetAddress of localhost");
		}

		socket.close();
		socket = null;

		try {
			clientServerSocket.close();
		} catch (IOException e1) {
			// ignore
		}
		clientServerSocket = null;

		// close all client connections
		synchronized (clients) {
			for (Client c : clients)
				try {
					c.socket.close();
				} catch (IOException e) {
					// ignore
				}
			clients.clear();
		}
	}

	private void parseBroadcastPacket(byte[] b, InetAddress address) {
		// avoid parsing packet send after calling stop()
		if (!running)
			return;

		Tree expr;
		try {
			expr = SExpressionParser.parse(b);
		} catch (TreeParserException e) {
			e.printStackTrace();
			return;
		}

		try {
			parseBroadcastMessage(expr.getAs(Message.class), address);
		} catch (DeserializableException e) {
			LOGGER.error("Can't deserialize message", e);
		}
	}

	private void parseBroadcastMessage(Message message, InetAddress address) {
		if (bindAddress.equals(address))
			// drop, only process if address == null
			return;

		if (address == null)
			address = bindAddress;

		LOGGER.trace("Remote: parsing message from " + address + ": " + message);
		MessageType type = message.getType();

		if (type == EVENT) {
			InfrastructureEvent event = message.getEvent();
			switch (event.getEventType()) {
			case HEARTBEAT:
			case MIGRATED:
			case NEW_MACHINE:
			case SHUT_DOWN_ANNOUNCEMENT:
				event.address = address;
				break;
			case ANSWER_DEVICE:
				Client client;
				synchronized (queries) {
					client = queries.get(message.getID());
				}
				if (client != null)
					sendClientMessage(Message.event(event), client);
				break;
			default:
				break;
			}
			sendClientMessage(Message.event(event));
		} else if (type == SEARCH) {
			for (PublishedDeviceDescription device : getDevices(message
					.getDevice())) {
				InfrastructureEvent answer = new InfrastructureEvent(
						ANSWER_DEVICE);
				answer.device = device;
				Message msg = Message.event(answer);
				msg.setID(message.getID());
				try {
					sendBroadcastMessage(msg);
				} catch (InfrastructureException e) {
					LOGGER.error("error sending answer", e);
				}
			}
		} else
			LOGGER.warn("Don't know how to handle message " + type);
	}

	private List<PublishedDeviceDescription> getDevices(
			PublishedDeviceDescription pattern) {
		LinkedList<PublishedDeviceDescription> result = new LinkedList<PublishedDeviceDescription>();

		synchronized (devices) {
			for (Pair<Client, PublishedDeviceDescription> device : devices)
				if (device.getSecond().matches(pattern))
					result.add(device.getSecond());
		}

		return result;
	}

	/**
	 * Add the device to the local list. If broadcast is true, notify everyone
	 * of this device
	 * 
	 * @param device
	 * @param broadcast
	 * 
	 * @throws InfrastructureException
	 *             when the message could not be sent.
	 */
	private void addDevice(PublishedDeviceDescription device, Client client,
			boolean broadcast) throws InfrastructureException {
		synchronized (devices) {
			devices.add(new Pair<Client, PublishedDeviceDescription>(client,
					device));
		}

		if (!broadcast)
			return;

		// notify everyone
		InfrastructureEvent event = new InfrastructureEvent(NEW_DEVICE);
		event.device = device;
		sendBroadcastMessage(Message.event(event));
	}

	/**
	 * Removes a single device
	 * 
	 * @param device
	 * @param client
	 * 
	 * @throws InfrastructureException
	 *             when the message could not be sent.
	 */
	private void removeDevice(PublishedDeviceDescription device, Client client,
			boolean clean) throws InfrastructureException {
		synchronized (devices) {
			devices.remove(new Pair<Client, PublishedDeviceDescription>(client,
					device));
		}

		sendRemoveDeviceMessage(device, clean);
	}

	/**
	 * removes all devices from the given client
	 * 
	 * @param client
	 * @param clean
	 *            how the device removal should be flagged
	 * 
	 * @throws InfrastructureException
	 *             when the message could not be sent.
	 */
	private void removeDevices(Client client, boolean clean)
			throws InfrastructureException {
		synchronized (devices) {
			Iterator<Pair<Client, PublishedDeviceDescription>> it = devices
					.iterator();
			while (it.hasNext()) {
				Pair<Client, PublishedDeviceDescription> device = it.next();
				if (device.getFirst().equals(client)) {
					it.remove();
					sendRemoveDeviceMessage(device.getSecond(), clean);
				}
			}
		}
	}

	/**
	 * @throws InfrastructureException
	 *             when the message could not be sent.
	 */
	private void sendRemoveDeviceMessage(PublishedDeviceDescription device,
			boolean clean) throws InfrastructureException {
		InfrastructureEvent event = new InfrastructureEvent(DEVICE_REMOVED);
		event.clean = clean;
		event.device = device;
		sendBroadcastMessage(Message.event(event));
	}

	/**
	 * @throws InfrastructureException
	 *             when the message could not be sent.
	 */
	private void searchDevice(PublishedDeviceDescription pattern, Tree id)
			throws InfrastructureException {
		sendBroadcastMessage(Message.searchDevice(pattern, id));
	}

	/**
	 * Send a broadcast message
	 * 
	 * @param message
	 * @throws InfrastructureException
	 *             when sending failed
	 */
	private void sendBroadcastMessage(Message message)
			throws InfrastructureException {
		LOGGER.trace("Remote: sending " + message);
		sendBroadcastMessage_(Tree.fromObject(message));

		// parse message as if we received it
		parseBroadcastMessage(message, null);
	}

	private void sendBroadcastMessage_(Tree message)
			throws InfrastructureException {
		byte[] b = SExpressionWriter.writeSExpression(message);

		DatagramPacket packet = new DatagramPacket(b, b.length,
				broadcastAddress, INFRASTRUCTURE_PORT);

		try {
			sendSocket.send(packet);
		} catch (IOException e) {
			throw new InfrastructureException(
					"Sending UDP infrastructure broadcast failed", e)
					.setType(InfrastructureException.Type.NETWORK_ERROR);
		}

	}

	/**
	 * parses the message received from a client
	 * 
	 * @param message
	 * 
	 * @throws InfrastructureException
	 */
	protected void parseClientMessage(Message message, Client client)
			throws InfrastructureException {
		LOGGER.trace("Parsing from local " + client + ": " + message);

		PublishedDeviceDescription device = message.getDevice();

		switch (message.getType()) {
		case ADD:
			addDevice(device, client, !message.isAddExisting());
			break;
		case REMOVE:
			removeDevice(device, client, true);
			break;
		case SEARCH:
			Tree ID = getNextUniqueQueryID(device);
			synchronized (queries) {
				queries.put(ID, client);
			}
			searchDevice(device, ID);
			break;
		case REBIND:
			// broadcast to all local clients
			sendClientMessage(message);
			break;
		default:
			LOGGER.warn("don't know how to handle message " + message
					+ " from client " + client);
		}
	}

	/**
	 * unique query IDs are needed to filter the responses of this query.
	 * 
	 * @return a probably unique ID to identify a query from this machine
	 */
	private Tree getNextUniqueQueryID(Object pattern) {
		int id = pattern.hashCode() + random.nextInt();
		LeafNode ID = new LeafNode(Integer.toHexString(id));
		return new InnerNode(machineID, ID);
	}

	/**
	 * Sends the message to all clients
	 * 
	 * @param message
	 */
	protected void sendClientMessage(Message message) {
		LOGGER.trace("Sending to all local: " + message);
		sendClientMessage_(Tree.fromObject(message));
	}

	/**
	 * Sends the message to the client
	 */
	protected void sendClientMessage(Message message, Client client) {
		LOGGER.trace("Sending to local " + client + ": " + message);
		sendClientMessage_(Tree.fromObject(message), client);
	}

	private void sendClientMessage_(Tree message) {
		byte[] b = SExpressionWriter.writeSExpression(message);
		synchronized (clients) {
			for (Client c : clients)
				sendClientMessage(c, b);
		}
	}

	private void sendClientMessage_(Tree message, Client client) {
		byte[] b = SExpressionWriter.writeSExpression(message);
		sendClientMessage(client, b);
	}

	/**
	 * Sends the message to the client
	 */
	private void sendClientMessage(Client client, byte[] b) {
		synchronized (client) {
			try {
				client.output.write(b);
				client.output.flush();
			} catch (IOException e) {
				LOGGER.error("Could not send message to client " + client, e);
				try {
					client.socket.close();
				} catch (IOException e1) {
					// ignore
				}
			}
		}
	}

	@Override
	protected void finalize() throws Throwable {
		stop();
		super.finalize();
	}
}
